package com.iwhalecloud.uncc.utils;

import com.ctg.mq.api.bean.MQResult;
import com.iwhalecloud.uncc.common.ArchiverConstant;
import com.iwhalecloud.uncc.common.SysConstant;
import com.iwhalecloud.uncc.db.DatasourceContextHolder;
import com.iwhalecloud.uncc.domain.ArchiverMsg;
import com.iwhalecloud.uncc.exception.ArchiverException;
import com.iwhalecloud.uncc.mq.handler.ArchiverHandler;
import com.iwhalecloud.uncc.mq.producer.MqProducerManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;

/**
 * @author zq
 * @date 2021/10/22 10:53
 * @description ArchiverUtil
 */
@Slf4j
public class ArchiverUtil {


    /**
     * 发送MQ消息 格式:Msg#UUID.randomUUID().toString()
     * @param topic
     * @param key
     * @param msg
     * @return
     * @throws Exception
     */
    public static boolean sendMsg(String topic,String key , ArchiverMsg msg) throws ArchiverException {
        log.info(Thread.currentThread().getName()+"发送MQ............");

        try {
            MqProducerManager.getProducer(topic).sendMsg(topic, SysConstant.MQ_DEFAULT_TAG, key,
                            JsonUtil.obj2Json(msg)+ArchiverConstant.MSG_SEPERATOR + UUID.randomUUID().toString(), true);
        }catch (MQBrokerException mqBrokerException){
            int responseCode = mqBrokerException.getResponseCode();
            mqBrokerException.printStackTrace();
            if(responseCode == 13){
                log.error("msg 大于 MQ最大消息限度 ， 需调小oneSendMax参数");
            }
            throw new ArchiverException("MQBrokerException异常");
        } catch (Exception e) {
            e.printStackTrace();
            throw new ArchiverException("MQ发送异常");
        }
        return true;
    }


    public static boolean dealMessage(List messages, Class<? extends ArchiverMsg> clazz , ArchiverHandler archiverHandler) {
        try {
            for (int i = 0; i < messages.size(); i++) {
                Object object = messages.get(i);
                if (object instanceof MessageExt) {
                    deal(((MessageExt) object).getBody(),clazz,archiverHandler);
                } else if (object instanceof MQResult) {
                    deal(((MQResult) object).getMessage().getBody(),clazz,archiverHandler);
                }
            }
        }catch (Exception e) {
            //回改数据
            e.printStackTrace();
        }finally {
            DatasourceContextHolder.setDs(ArchiverConstant.DB_DEFAULT_HIS);
        }
        return true;
    }

    private static void   deal(byte[] body, Class<? extends ArchiverMsg> clazz ,ArchiverHandler archiverHandler) {
        String message = new String(ZipUtil.uzip(body), StandardCharsets.UTF_8);
        String[] strings = message.split(ArchiverConstant.MSG_SEPERATOR);
        ArchiverMsg msg = (ArchiverMsg) JsonUtil.json2Obj(strings[0], clazz);
        //处理msg
        archiverHandler.dealArchiverMsg(msg);
    }
}
